{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyalink.alink import *\n",
    "useLocalEnv(1)\n",
    "\n",
    "from utils import *\n",
    "import os\n",
    "import pandas as pd\n",
    "\n",
    "DATA_DIR = ROOT_DIR + \"temp\" + os.sep"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_5_1\n",
    "source = CsvSourceBatchOp()\\\n",
    "    .setFilePath(\"http://archive.ics.uci.edu/ml/machine-learning-databases\"\n",
    "                 + \"/iris/iris.data\")\\\n",
    "    .setSchemaStr(\"sepal_length double, sepal_width double, petal_length double, \"\n",
    "                  + \"petal_width double, category string\")\n",
    "\n",
    "source.firstN(5).print();\n",
    "\n",
    "source.sampleWithSize(10)\\\n",
    "    .link(\n",
    "        CsvSinkBatchOp()\\\n",
    "            .setFilePath(DATA_DIR + \"iris_10.data\")\\\n",
    "            .setOverwriteSink(True)\n",
    "    )\n",
    "\n",
    "BatchOperator.execute();"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_5_2\n",
    "df = pd.DataFrame(\n",
    "    [\n",
    "        [2009, 0.5],\n",
    "        [2010, 9.36],\n",
    "        [2011, 52.0],\n",
    "        [2012, 191.0],\n",
    "        [2013, 350.0],\n",
    "        [2014, 571.0],\n",
    "        [2015, 912.0],\n",
    "        [2016, 1207.0],\n",
    "        [2017, 1682.0]\n",
    "    ]\n",
    ")  \n",
    "train_set = BatchOperator.fromDataframe(df, schemaStr='x int, gmv double')\n",
    "\n",
    "df_2 = pd.DataFrame(\n",
    "    [\n",
    "        [2018],\n",
    "        [2019]\n",
    "    ]\n",
    ") \n",
    "pred_set = BatchOperator.fromDataframe(df_2, schemaStr='x int')\n",
    "\n",
    "train_set = train_set.select(\"x, x*x AS x2, gmv\");\n",
    "\n",
    "trainer = LinearRegTrainBatchOp()\\\n",
    "    .setFeatureCols([\"x\", \"x2\"])\\\n",
    "    .setLabelCol(\"gmv\")\n",
    "\n",
    "train_set.link(trainer);\n",
    "\n",
    "trainer.link(\n",
    "    AkSinkBatchOp()\\\n",
    "        .setFilePath(DATA_DIR + \"gmv_reg.model\")\\\n",
    "        .setOverwriteSink(True)\n",
    ")\n",
    "BatchOperator.execute()\n",
    "\n",
    "lr_model = AkSourceBatchOp().setFilePath(DATA_DIR + \"gmv_reg.model\")\n",
    "\n",
    "pred_set = pred_set.select(\"x, x*x AS x2\")\n",
    "\n",
    "predictor = LinearRegPredictBatchOp().setPredictionCol(\"pred\")\n",
    "\n",
    "predictor.linkFrom(lr_model, pred_set).print();"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_5_3\n",
    "pred_set = StreamOperator.fromDataframe(df_2, schemaStr='x int')\n",
    "\n",
    "lr_model = AkSourceBatchOp().setFilePath(DATA_DIR + \"gmv_reg.model\")\n",
    "\n",
    "predictor = LinearRegPredictStreamOp(lr_model).setPredictionCol(\"pred\")\n",
    "\n",
    "pred_set\\\n",
    "    .select(\"x, x*x AS x2\")\\\n",
    "    .link(predictor)\\\n",
    "    .print()\n",
    "\n",
    "StreamOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_5_4\n",
    "if os.path.exists(DATA_DIR + \"gmv_pipeline.model\"):\n",
    "    os.remove(DATA_DIR + \"gmv_pipeline.model\")\n",
    "\n",
    "train_set = BatchOperator.fromDataframe(df, schemaStr='x int, gmv double')\n",
    "\n",
    "pipeline = Pipeline()\\\n",
    "    .add(\n",
    "        Select().setClause(\"*, x*x AS x2\")\n",
    "    )\\\n",
    "    .add(\n",
    "        LinearRegression()\\\n",
    "            .setFeatureCols([\"x\", \"x2\"])\\\n",
    "            .setLabelCol(\"gmv\")\\\n",
    "            .setPredictionCol(\"pred\")\n",
    "    )\n",
    "\n",
    "pipeline.fit(train_set).save(DATA_DIR + \"gmv_pipeline.model\")\n",
    "\n",
    "BatchOperator.execute()\n",
    "\n",
    "pipelineModel = PipelineModel.load(DATA_DIR + \"gmv_pipeline.model\")\n",
    "\n",
    "pred_batch = BatchOperator.fromDataframe(df_2, schemaStr='x int')\n",
    "\n",
    "pipelineModel.transform(pred_batch).print()\n",
    "\n",
    "\n",
    "pred_stream = StreamOperator.fromDataframe(df_2, schemaStr='x int')\n",
    "\n",
    "pipelineModel.transform(pred_stream).print()\n",
    "\n",
    "StreamOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_5_5\n",
    "\n",
    "predictor = LocalPredictor(DATA_DIR + \"gmv_pipeline.model\", \"x int\")\n",
    "\n",
    "print(predictor.getOutputColNames())\n",
    "\n",
    "for x in [2018, 2019] :\n",
    "    print(predictor.map([x]))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
